/*
 * @Author: wuqingchun
 * @Date: 2024-07-01 16:10:07
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-07-08 20:26:01
 */

#include <sys/prctl.h>

#include "arq_receiver.h"

CARQReceiver::CARQReceiver():
    CARQSocket(1)
{
    
}

int32_t CARQReceiver::Init(SOCKADDR_UN& refPeerAddr, uint32_t uiWndSize)
{
    int32_t iErr = m_objRcvWnd.Init(uiWndSize);
    if (iErr == 0)
    {
        iErr = CARQSocket::Init(refPeerAddr);
    }

    return iErr;
}

int32_t CARQReceiver::Start()
{
    m_bRecvRuning = true;
    m_objRecvThread = std::thread(&CARQReceiver::__ThreadFuncRecv, this);

    return 0;
}

void CARQReceiver::GetStatistics()
{
    m_objRcvWnd.GetStatistics();
}

void CARQReceiver::__ThreadFuncRecv()
{
    ssize_t lRes;
    size_t ulBufLen = 1024;

    (void)prctl(PR_SET_NAME, "rx_recv");

    while (m_bRecvRuning)
    {
        void* pBuf = m_objAllocator.Alloc(ulBufLen);
        if (pBuf == nullptr)
        {
            break;
        }

        lRes = CARQSocket::Recv(pBuf, ulBufLen);
        if (lRes >= sizeof(ARQ_HEAD_S))
        {
            __HandleMsg(*(ARQ_HEAD_S*)pBuf);
        }
    }
}

int32_t CARQReceiver::__HandleMsg(ARQ_HEAD_S& refHead)
{
    if (m_objRcvWnd.OnRecv(&refHead))
    {
        // 收到连续的消息，发送ACK
        return __SendAck(m_objRcvWnd.GetNextSeqNo() - 1);
    }

    return -1;
}

int32_t CARQReceiver::__SendAck(uint32_t uiAckNo)
{
    ARQ_HEAD_S stAck;

    stAck.uiSeqNo = 0;
    stAck.uiAckNo = uiAckNo;

    return CARQSocket::Send(&stAck, sizeof(ARQ_HEAD_S));
}